///////////////////////////////////////////////////////////////////////////////
// File:        server.c
// Author:      671643387@qq.com
// Description:
///////////////////////////////////////////////////////////////////////////////

#include <assert.h>
#include <errno.h>
#include <memory.h>
#include <malloc.h>
#include <signal.h>
#include <limits.h>
#if defined(__WINDOWS__)
#include <process.h>
#elif defined(__LINUX__)
#include <unistd.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <pthread.h>
#endif
#include <event2/event.h>
#include "tstl2cl/c_map.h"
#include "pio/server.h"

///////////////////////////////////////////////////////////////////////////////

enum pio_server_state
{
    PIO_SERVER_NOT_READY = 1,
    PIO_SERVER_READY,
    PIO_SERVER_RUNNING,
    PIO_SERVER_EXITING,
    PIO_SERVER_EXIT
};

struct pio_connection
{
    int                     id;
    evutil_socket_t         sock;
    struct event*           eread;
    struct event*           ewrite;
    struct pio_server*      server;
    struct pio_worker*      worker;

    connection_event_func   read_func;
    void*                   read_arg;
    connection_event_func   write_func;
    void*                   write_arg;
    connection_event_func   close_func;
    void*                   close_arg;

    void*                   user_data;

    c_pair                  pair;
};

struct pio_server
{
    const char*             address;
    short                   port;
    int                     backlog;
    evutil_socket_t         slisten;
    struct event*           elisten;

    struct event_base*      mbase;

    int                     id_gen;
    int                     conn_total;
    int                     conn_load;
    enum pio_server_state   state;

    int                     worker;
    struct pio_worker**     workers;

    connection_create_func  conn_create_func;
    connection_free_func    conn_free_func;

    connection_event_func   conn_establised_func;
    void*                   conn_establised_arg;

    connection_event_func   conn_read_func;
    void*                   conn_read_arg;
    connection_event_func   conn_write_func;
    void*                   conn_write_arg;
    connection_event_func   conn_close_func;
    void*                   conn_close_arg;

    c_map                   conn_map;
#if defined(__WINDOWS__)
    CRITICAL_SECTION        conn_map_lock;
#elif defined(__LINUX__)
    pthread_mutex_t         conn_map_lock;
#endif
};

struct pio_worker
{
    int                     conn_load;
    struct event_base*      base;
    struct event*           dummy;

#if defined(__WINDOWS__)
    uintptr_t               handle;
    DWORD                   id;
#elif defined(__LINUX__)
    pthread_t               id;
#endif
};

///////////////////////////////////////////////////////////////////////////////

static int int_comparer(void * x, void * y)
{
    return *(int *)(x) - *(int *)(y);
}

///////////////////////////////////////////////////////////////////////////////

static int closesock(evutil_socket_t sock)
{
#if defined(__WINDOWS__)
    closesocket(sock);
#elif defined(__LINUX__)
    close(sock);
#endif
    return 0;
}

static int setnonblocking(evutil_socket_t sock)
{
#if defined(__WINDOWS__)
    u_long arg = 1;
    return ioctlsocket(sock, FIONBIO, &arg);
#elif defined(__LINUX__)
    int flags = fcntl(sock, F_GETFL, 0);
    flags |= O_NONBLOCK;
    return (-1 != fcntl(sock, F_SETFL, flags)) ? 0 : -1;
#endif
}

static int inc32(int* val)
{
#if defined(__WINDOWS__)
    return (int)InterlockedIncrement((long*)val);
#elif defined(__LINUX__)
    return (int)__sync_add_and_fetch(val, 1);
#endif
}

static int dec32(int* val)
{
#if defined(__WINDOWS__)
    return (int)InterlockedDecrement((long*)val);
#elif defined(__LINUX__)
    return (int)__sync_sub_and_fetch(val, 1);
#endif
}

#if defined(__WINDOWS__)
static unsigned int __stdcall start_address(void* arg);
#elif defined(__LINUX__)
static void* start_address(void* arg);
#endif

///////////////////////////////////////////////////////////////////////////////

static struct pio_worker* pio_worker_new(void);
static void pio_worker_stop(struct pio_worker*);
static void pio_worker_free(struct pio_worker*);
static int pio_worker_run(struct pio_worker*);

///////////////////////////////////////////////////////////////////////////////

static void do_pio_connection_close(struct pio_connection*);

static int find_fit_base(struct pio_server* serv)
{
    int x, i = 0;
    int min_load = INT_MAX;

    assert(serv && serv->state == PIO_SERVER_RUNNING);

    if (serv->worker == 0)
        return -1;

    for (x = 0; x < serv->worker; ++x)
    {
        if (serv->workers[x]->conn_load < min_load)
        {
            min_load = serv->workers[x]->conn_load;
            i = x;
        }
    }

    if (serv->conn_load < min_load)
        return -1;
    return i;
}

static int do_pio_server_start_workers(struct pio_server* serv)
{
    int x;
    assert(serv);

    if (serv->worker == 0)
        return 0;

    for (x = 0; x < serv->worker; ++x)
    {
        if (0 != pio_worker_run(serv->workers[x]))
            return -1;
    }

    return 0;
}

static void do_pio_server_stop_workers(struct pio_server* serv)
{
    int x;
    assert(serv);
    if (serv->worker == 0)
        return;
    for (x = 0; x < serv->worker; ++x)
        pio_worker_stop(serv->workers[x]);
}

static void do_pio_server_free_workers(struct pio_server* serv)
{
    int x;
    assert(serv);
    if (serv->worker == 0)
        return;
    for (x = 0; x < serv->worker; ++x)
        pio_worker_free(serv->workers[x]);
    free(serv->workers);
    serv->workers = NULL;
    serv->worker = 0;
}

///////////////////////////////////////////////////////////////////////////////

static void on_accept(evutil_socket_t sock, short event, void* arg);
static void on_read(evutil_socket_t sock, short event, void* arg);
static void on_write(evutil_socket_t sock, short event, void* arg);
static void on_timeout(evutil_socket_t sock, short event, void* arg);

///////////////////////////////////////////////////////////////////////////////

struct pio_server* pio_server_new()
{
    struct pio_server* newserv = 
        (struct pio_server*)malloc(sizeof(struct pio_server));
    if (newserv)
    {
        memset(newserv, 0, sizeof(struct pio_server));

        do 
        {
            newserv->mbase = event_base_new();
            if (!newserv->mbase) break;

            newserv->slisten = socket(AF_INET, SOCK_STREAM, 0);
            if (newserv->slisten < 0) break;

            newserv->elisten = event_new(newserv->mbase, 
                newserv->slisten, EV_READ|EV_PERSIST, on_accept, newserv);
            if (!newserv->elisten) break;

            c_map_create(&newserv->conn_map, int_comparer);

            event_base_set(newserv->mbase, newserv->elisten);
#if defined(__WINDOWS__)
            InitializeCriticalSection(&(newserv->conn_map_lock));
#elif defined(__LINUX__)
			pthread_mutex_init(&(newserv->conn_map_lock), NULL);
#endif
            newserv->state = PIO_SERVER_NOT_READY;
        } while (0);

        if (newserv && newserv->state != PIO_SERVER_NOT_READY)
        {
            pio_server_free(newserv);
            newserv = NULL;
        }
    }
    return newserv;
}

int pio_server_open(struct pio_server* serv, const char* address, short port, int backlog)
{
    struct sockaddr_in saddr;

    assert(serv && serv->state == PIO_SERVER_NOT_READY && address);

    memset(&saddr, 0, sizeof(saddr));
    saddr.sin_family = AF_INET;
#if defined(__WINDOWS__)
    saddr.sin_addr.S_un.S_addr = inet_addr(address);
#elif defined(__LINUX__)
    saddr.sin_addr.s_addr = inet_addr(address);
#endif
    saddr.sin_port = htons(port);

    if (bind(serv->slisten, (const struct sockaddr*)&saddr, sizeof(saddr)) < 0)
        return -1;

    if (setnonblocking(serv->slisten) < 0)
        return -1;

    if (listen(serv->slisten, backlog) < 0)
        return -1;

    if (0 != event_add(serv->elisten, NULL))
        return -1;

    serv->address = address;
    serv->port = port;
    serv->backlog = backlog;

    if (0 != do_pio_server_start_workers(serv))
    {
        fprintf(stderr, "do_pio_server_start_workers failed: %d\n", errno);
        return -1;
    }

#if defined(__LINUX__)
    signal(SIGPIPE, SIG_IGN);
#endif
    serv->state = PIO_SERVER_READY;
    return 0;
}

int pio_server_run(struct pio_server* serv)
{
    struct pio_connection* conn;
    c_iterator iter, end;

    assert(serv && serv->state == PIO_SERVER_READY);
    serv->state = PIO_SERVER_RUNNING;
    event_base_dispatch(serv->mbase);

    ///////////////////////////////////////////////////////////////////////////

#if defined(__WINDOWS__)
    EnterCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
    pthread_mutex_lock(&(serv->conn_map_lock));
#endif

    // close all connections
    iter = c_map_begin(&(serv->conn_map));
    end = c_map_end(&(serv->conn_map));
    while(!ITER_EQUAL(iter, end))
    {
        c_iterator to_erase = iter;
        conn = (struct pio_connection*)((c_ppair)ITER_REF(iter))->second;
        c_map_erase(&(serv->conn_map), to_erase);
        do_pio_connection_close(conn);

        assert(__c_rb_tree_verify((&serv->conn_map)->_l));
        iter = c_map_begin(&(serv->conn_map));
        end = c_map_end(&(serv->conn_map));
    }
    assert(c_map_empty(&serv->conn_map));
    serv->conn_total = 0;
    serv->conn_load = 0;

#if defined(__WINDOWS__)
    LeaveCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
    pthread_mutex_unlock(&(serv->conn_map_lock));
#endif

    ///////////////////////////////////////////////////////////////////////////

    // free workers
    do_pio_server_free_workers(serv);

    serv->state = PIO_SERVER_EXIT;
    return 0;
}

void pio_server_stop(struct pio_server* serv)
{
    assert(serv && serv->state == PIO_SERVER_RUNNING);
    serv->state = PIO_SERVER_EXITING;
    
    // close listener
    event_free(serv->elisten);
    serv->elisten = NULL;
    closesock(serv->slisten);
    serv->slisten = -1;

    // stop worker
    do_pio_server_stop_workers(serv);

    // loop break
    event_base_loopbreak(serv->mbase);
}

void pio_server_free(struct pio_server* serv)
{
    assert(serv);

#if defined(__WINDOWS__)
    DeleteCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
    pthread_mutex_destroy(&(serv->conn_map_lock));
#endif

    c_map_destroy(&serv->conn_map);

    if (serv->elisten)
    {
        event_free(serv->elisten);
        serv->elisten = NULL;
    }

    if (serv->slisten > 0)
    {
        closesock(serv->slisten);
        serv->slisten = -1;
    }

    if (serv->mbase)
    {
        event_base_free(serv->mbase);
        serv->mbase = NULL;
    }
    free(serv);
}

void pio_server_set_connection_create_fn(struct pio_server* serv, connection_create_func fn)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_create_func = fn;
}

void pio_server_set_connection_free_fn(struct pio_server* serv, connection_free_func fn)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_free_func = fn;
}

void pio_server_set_connection_establised_fn(struct pio_server* serv, connection_event_func fn, void* arg)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_establised_func = fn;
    serv->conn_establised_arg = arg;
}

void pio_server_set_connection_read_fn(struct pio_server* serv, connection_event_func fn, void* arg)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_read_func = fn;
    serv->conn_read_arg = arg;
}

void pio_server_set_connection_write_fn(struct pio_server* serv, connection_event_func fn, void* arg)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_write_func = fn;
    serv->conn_write_arg = arg;
}

void pio_server_set_connection_close_fn(struct pio_server* serv, connection_event_func fn, void* arg)
{
    assert(serv && serv->state == PIO_SERVER_NOT_READY && fn);
    serv->conn_close_func = fn;
    serv->conn_close_arg = arg;
}

int pio_server_config_worker(struct pio_server* serv, int worker)
{
    int x, z = 0;
    assert(serv && serv->state == PIO_SERVER_NOT_READY && serv->worker == 0);
    if (worker <= 0) 
        return 0;
    serv->worker = worker;
    serv->workers = (struct pio_worker**)malloc(sizeof(struct pio_worker*) * worker);
    if (serv->workers)
    {
        do 
        {
            for (x = 0; x < worker; ++x)
            {
                serv->workers[x] = pio_worker_new();
                if (!serv->workers[x])
                    break;
                ++z;
            }
        } while (0);

        if (z != worker)
        {
            for (x = 0; x < worker; ++x)
            {
                if (serv->workers[x])
                {
                    pio_worker_free(serv->workers[x]);
                }
            }

            free(serv->workers);
            serv->workers = NULL;
            serv->worker = 0;
        }

        return (z == worker) ? 0 : -1;
    }

    serv->worker = 0;
    return -1;
}

struct pio_worker* pio_server_get_worker(struct pio_server* serv, int index)
{
    assert(serv);
    if (serv->worker > 0 && index >= 0 && index < serv->worker)
        return serv->workers[index];
    return NULL;
}

void pio_server_get_statics(struct pio_server* serv, struct pio_server_statics* sta)
{
    assert(serv && sta);
    sta->address = serv->address;
    sta->port = serv->port;
    sta->backlog = serv->backlog;
    sta->id_gen = serv->id_gen;
    sta->conn_total = serv->conn_total;
    sta->conn_load = serv->conn_load;
    sta->worker = serv->worker;
}

///////////////////////////////////////////////////////////////////////////////

int pio_connectoon_size(void)
{
    return sizeof(struct pio_connection);
}

int pio_connection_get_id(struct pio_connection* conn)
{
    assert(conn);
    return conn->id;
}

int pio_connection_notify_write(struct pio_connection* conn)
{
    struct event_base* base = 
        conn->worker ? conn->worker->base : conn->server->mbase;

    assert(conn);
    if (!conn->ewrite)
    {
        conn->ewrite = event_new(base, 
            conn->sock, EV_WRITE, on_write, conn);
    }

    if (!conn->ewrite)
        return -1;

    event_base_set(base, conn->ewrite);
    return event_add(conn->ewrite, NULL);
}

int pio_connection_read(struct pio_connection* conn, char* buff, unsigned int length, int flags)
{
    int ret = 0;
#if defined(__WINDOWS__)
    int wsaerr = 0;
#endif

    assert(conn && buff);

#if defined(__WINDOWS__)
    ret = recv(conn->sock, buff, length, flags);
#elif defined(__LINUX__)
    ret = recv(conn->sock, buff, length, flags);
#endif
    if (ret == SOCKET_ERROR)
    {
#if defined(__LINUX__)
        if (errno == EWOULDBLOCK)
        {
            return SOCKET_ERROR_WOULDBLOCK;
        }
#elif defined(__WINDOWS__)
        wsaerr = WSAGetLastError();
        if (wsaerr == WSAEWOULDBLOCK)
        {
            return SOCKET_ERROR_WOULDBLOCK;
        }
#endif
    }
    else if (ret == 0)
    {
    }
    return ret;
}

int pio_connection_write(struct pio_connection* conn, const char* buff, unsigned int length, int flags)
{
    int ret = 0;
#if defined(__WINDOWS__)
    int wsaerr = 0;
#endif

    assert(conn && buff);

#if defined(__WINDOWS__)
    ret = send(conn->sock, buff, length, flags);
#elif defined(__LINUX__)
    ret = send(conn->sock, buff, length, flags);
#endif
    if (ret == SOCKET_ERROR)
    {
#if defined(__LINUX__)
        if (errno == EWOULDBLOCK)
        {
            return SOCKET_ERROR_WOULDBLOCK;
        }
#elif defined(__WINDOWS__)
        wsaerr = WSAGetLastError();
        if (wsaerr == WSAEWOULDBLOCK)
        {
            return SOCKET_ERROR_WOULDBLOCK;
        }
#endif
    }
    return ret;
}

void pio_connection_close(struct pio_connection* conn)
{
    struct pio_server* serv = conn->server;
    c_iterator find, map_end;

    assert(conn && serv);
    if (conn->eread)
    {
        event_free(conn->eread);
        conn->eread = NULL;
    }

    if (conn->ewrite)
    {
        event_free(conn->ewrite);
        conn->ewrite = NULL;
    }

    conn->close_func(conn, conn->close_arg);

    if (conn->worker)
        dec32(&(conn->worker->conn_load));
    else
        dec32(&(conn->server->conn_load));

    ///////////////////////////////////////////////////////////////////////////

#if defined(__WINDOWS__)
    EnterCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
    pthread_mutex_lock(&(serv->conn_map_lock));
#endif

    find = c_map_find(&serv->conn_map, &conn->id);
    map_end = c_map_end(&serv->conn_map);
    if(!ITER_EQUAL(map_end, find))
    {
        c_map_erase(&serv->conn_map, find);
        dec32(&(serv->conn_total));
    }

#if defined(__WINDOWS__)
    LeaveCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
    pthread_mutex_unlock(&(serv->conn_map_lock));
#endif

    ///////////////////////////////////////////////////////////////////////////

    if (serv->conn_free_func)
        serv->conn_free_func(conn);
    else
        free(conn);
}

void do_pio_connection_close(struct pio_connection* conn)
{
    struct pio_server* serv = conn->server;
    assert(conn && serv);
    if (conn->eread)
    {
        event_free(conn->eread);
        conn->eread = NULL;
    }

    if (conn->ewrite)
    {
        event_free(conn->ewrite);
        conn->ewrite = NULL;
    }

    conn->close_func(conn, conn->close_arg);

    if (conn->worker)
        dec32(&(conn->worker->conn_load));
    else
        dec32(&(conn->server->conn_load));

    if (serv->conn_free_func)
        serv->conn_free_func(conn);
    else
        free(conn);
}

void pio_connection_set_userdata(struct pio_connection* conn, void* data)
{
    assert(conn);
    conn->user_data = data;
}

void* pio_connection_get_userdata(struct pio_connection* conn)
{
    assert(conn);
    return conn->user_data;
}

void pio_connection_set_read_fn(struct pio_connection* conn, connection_event_func fn, void* arg)
{
    assert(conn && fn);
    conn->read_func = fn;
    conn->read_arg = arg;
}

void pio_connection_set_write_fn(struct pio_connection* conn, connection_event_func fn, void* arg)
{
    assert(conn && fn);
    conn->write_func = fn;
    conn->write_arg = arg;
}

void pio_connection_set_close_fn(struct pio_connection* conn, connection_event_func fn, void* arg)
{
    assert(conn && fn);
    conn->close_func = fn;
    conn->close_arg = arg;
}

struct pio_server* pio_connection_get_server(struct pio_connection* conn)
{
    assert(conn);
    return conn->server;
}

///////////////////////////////////////////////////////////////////////////////

struct pio_worker* pio_worker_new(void)
{
    struct timeval tv = {1, 0};
    struct pio_worker* worker = (struct pio_worker*)malloc(sizeof(struct pio_worker));
    if (!worker)
        return NULL;

    memset(worker, 0, sizeof(struct pio_worker));
    do 
    {
        worker->base = event_base_new();
        if (!worker->base) break;

        worker->dummy = event_new(worker->base, -1, 
            EV_TIMEOUT | EV_PERSIST, on_timeout, worker);
        if (!worker->dummy) break;

        event_base_set(worker->base, worker->dummy);
        if (0 != event_add(worker->dummy, &tv))
            break;

#if defined(__WINDOWS__)
        worker->handle = -1L;
#endif

        return worker;
    } while (0);

    if (worker->dummy)
        event_free(worker->dummy);
    if (worker->base)
        event_base_free(worker->base);

    free(worker);
    return NULL;
}

void pio_worker_stop(struct pio_worker* worker)
{
    assert(worker);
    event_base_loopbreak(worker->base);
#if defined(__WINDOWS__)
    assert(worker->handle != -1L);
    WaitForSingleObject((HANDLE)(worker->handle), INFINITE);
    CloseHandle((HANDLE)(worker->handle));
    worker->handle = -1L;
#elif defined(__LINUX__)
    pthread_join(worker->id, NULL);
#endif
}

void pio_worker_free(struct pio_worker* worker)
{
    assert(worker);
    if (worker->dummy)
        event_free(worker->dummy);
    if (worker->base)
        event_base_free(worker->base);
    free(worker);
}

int pio_worker_run(struct pio_worker* worker)
{
    assert(worker);
#if defined(__WINDOWS__)
    worker->handle = _beginthreadex(NULL, 0, 
        start_address, worker, 0, &(worker->id));
    return (worker->handle != -1L) ? 0 : -1;
#elif defined(__LINUX__)
    return pthread_create(&(worker->id), NULL, start_address, worker);
#endif
}

void pio_worker_get_statics(struct pio_worker* worker, struct pio_worker_statics* sta)
{
    assert(worker && sta);
    sta->conn_load = worker->conn_load;
    sta->id = worker->id;
}

///////////////////////////////////////////////////////////////////////////////

void on_accept(evutil_socket_t sock, short event, void* arg)
{
    int ibase = 0;
    struct event_base* sbase = NULL;
    evutil_socket_t newsock = -1;
    struct pio_connection* newconn = NULL;
    struct pio_server* serv = (struct pio_server*)arg;
    c_iterator find, map_end;

    assert(serv);
    newsock = accept(serv->slisten, NULL, NULL);
    if (newsock < 0)
        return;

    if (setnonblocking(newsock) < 0)
    {
        closesock(newsock);
        return;
    }

    if (serv->conn_create_func)
        newconn = serv->conn_create_func();
    else
    {
        newconn = (struct pio_connection*)malloc(sizeof(struct pio_connection));
        memset(newconn, 0, sizeof(struct pio_connection));
    }
    

    if (newconn)
    {
        do 
        {
            ///////////////////////////////////////////////////////////////////
            ibase = find_fit_base(serv);
            if (ibase < 0)
                sbase = serv->mbase;
            else
                sbase = serv->workers[ibase]->base;

            newconn->eread = event_new(sbase, newsock, 
                EV_READ | EV_PERSIST, on_read, newconn);
            if (!newconn->eread)
            {
                fprintf(stderr, "create read event failed.\n");
                break;
            }

            event_base_set(sbase, newconn->eread);

            if (ibase >= 0)
            {
                inc32(&(serv->workers[ibase]->conn_load));
                newconn->worker = serv->workers[ibase];
            }
            else
                inc32(&(serv->conn_load));

            ///////////////////////////////////////////////////////////////////

#if defined(__WINDOWS__)
            EnterCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
            pthread_mutex_lock(&(serv->conn_map_lock));
#endif

            newconn->id = ++serv->id_gen;
            find = c_map_find(&serv->conn_map, &newconn->id);
            map_end = c_map_end(&serv->conn_map);
            if(!ITER_EQUAL(map_end, find))
            {
#if defined(__WINDOWS__)
                LeaveCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
                pthread_mutex_unlock(&(serv->conn_map_lock));
#endif
                fprintf(stderr, "connection id [%d] in use.\n", newconn->id);
                break;
            }

            newconn->server = serv;
            newconn->sock = newsock;
            newconn->read_func = serv->conn_read_func;
            newconn->write_func = serv->conn_write_func;
            newconn->close_func = serv->conn_close_func;
            newconn->read_arg = serv->conn_read_arg;
            newconn->write_arg = serv->conn_write_arg;
            newconn->close_arg = serv->conn_close_arg;

            newconn->pair.first = &newconn->id;
            newconn->pair.second = newconn;
            c_map_insert(&serv->conn_map, &newconn->pair);
            inc32(&(serv->conn_total));

#if defined(__WINDOWS__)
            LeaveCriticalSection(&(serv->conn_map_lock));
#elif defined(__LINUX__)
            pthread_mutex_unlock(&(serv->conn_map_lock));
#endif

            ///////////////////////////////////////////////////////////////////

            serv->conn_establised_func(newconn, serv->conn_establised_arg);
            if (0 != event_add(newconn->eread, NULL))
            {
                fprintf(stderr, "event_add for connection read failed.\n");
                pio_connection_close(newconn);
            }
            return;
        } while (false);

        closesock(newsock);
        if (serv->conn_free_func)
            serv->conn_free_func(newconn);
        else
            free(newconn);
    }
    else
        closesock(newsock);
}

void on_read(evutil_socket_t sock, short event, void* arg)
{
    struct pio_connection* conn = (struct pio_connection*)arg;
    assert(conn && conn->read_func);
    conn->read_func(conn, conn->read_arg);
}

void on_write(evutil_socket_t sock, short event, void* arg)
{
    struct pio_connection* conn = (struct pio_connection*)arg;
    assert(conn && conn->write_func);
    conn->write_func(conn, conn->write_arg);
}

///////////////////////////////////////////////////////////////////////////////

#if defined(__WINDOWS__)
unsigned int __stdcall start_address(void* arg)
#elif defined(__LINUX__)
void* start_address(void* arg)
#endif
{
    struct pio_worker* worker = (struct pio_worker*)arg;
    assert(worker);
    event_base_dispatch(worker->base);
    return 0;
}

void on_timeout(evutil_socket_t sock, short event, void* arg)
{

}
